/**
 * @author zhounan 
 * @date 2024年6月5日
 * @version V1.0
 */
package org.zn.etl.job.handler;

import static org.zn.etl.job.context.GlobalContextKey.END_OFFSET;
import static org.zn.etl.job.context.GlobalContextKey.ERROR_COUNT;
import static org.zn.etl.job.context.GlobalContextKey.ERROR_THRESHOLD;
import static org.zn.etl.job.context.GlobalContextKey.SRC_PK;
import static org.zn.etl.job.context.GlobalContextKey.START_OFFSET;
import static org.zn.etl.job.context.GlobalContextKey.TOTALROWS;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.stream.Collectors;

import javax.sql.DataSource;

import org.apache.commons.lang3.tuple.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.jdbc.core.simple.SimpleJdbcInsert;
import org.zn.datax.common.entity.ColumnInfo;
import org.zn.etl.jdbc.Upsert;
import org.zn.etl.misc.ErrorMessageFilterHelper;

import com.alibaba.fastjson.JSON;

import lombok.extern.slf4j.Slf4j;

/**
 * @author zhounan 
 * @date 2024年6月5日
 * @version V1.0
 */
@Slf4j
public class WriteAgent {
	
		private Upsert upsert;
		private SimpleJdbcInsert insert;
		
		private String  writeType;
		private String schema;
		private String tableName ;
		private String pk;
		private List<String> columns;
		
		private JdbcTemplate template;
		
		private List<Pair<ColumnInfo, ColumnInfo>> columnMappingPairs;
		
		private  Map<String, String>  tName2sName;
		
		private static final String MODE_UPSERT="upsert";
		private static final String MODE_INSERT="insert";
		
		private Logger writeErrorLogger =LoggerFactory.getLogger("etl.job.proto.error.data");
		
		/**
		 * @param writeType
		 * @param schema
		 * @param tableName
		 * @param pk
		 * @param columns
		 */
		public WriteAgent(String writeType, String schema, String tableName, String pk, List<Pair<ColumnInfo, ColumnInfo>> columnMappingPairs,DataSource ds,String dbType) {
			super();
			this.writeType = writeType!=null?writeType:MODE_INSERT;
			this.schema = schema;
			this.tableName = tableName;
			this.pk = pk;
			this.columnMappingPairs = columnMappingPairs;
			
			columns=columnMappingPairs.stream().map( it-> it.getRight().getName() ).collect(Collectors.toList());
			
			template=new JdbcTemplate(ds);
			
			if(MODE_INSERT.equals(this.writeType)) {
				insert=new SimpleJdbcInsert(template);
				insert.setTableName(schema+"."+tableName);
				insert.setCatalogName(schema);
				insert.setColumnNames(columns);
			}
			else if(MODE_UPSERT.equals(this.writeType)) {
				
				tName2sName= columnMappingPairs.stream().collect(Collectors.toMap(k-> k.getRight().getName(), v-> v.getLeft().getName()));
				upsert=Upsert.newUpsert(dbType, template).set(schema, tableName, pk, columns);
				
			}
			
		}
		
		
		public int execute(List<Map<String, Object>> list,List<Pair<ColumnInfo, ColumnInfo>> columnMappingPairs,Map<String, Object> context) throws WriteErrorOverException {
			
			int writed=0;
			
			if(MODE_INSERT.equals(writeType)) {
				
				writed=batchInsert(list, context);
			}
			else if(MODE_UPSERT.equals(writeType)) {
				
				writed=batchUpsert(list, context);
			}
			return writed;
			
		}
		
		private int batchUpsert(List<Map<String, Object>> list,Map<String, Object> context) throws WriteErrorOverException {
			
			List<Object[]> values=list.stream().map(it->{

				Object[] vs=new Object[columns.size()*2-1];
				int i=0;
				int x=0;
				
				for (String col : columns) {
					
					vs[i++]=it.get( tName2sName.get(col).toLowerCase());
					if(!Objects.equals(pk, col)) {
						vs[columns.size()+x]=it.get( tName2sName.get(col).toLowerCase() );
						x++;
					}
					
				}
				return vs;
			}).collect(Collectors.toList());
			
			int w=values.size();
			
			try {
				upsert.batchUpdateDirect(values);
			}catch (Exception e) {
				log.error(" etl write error,data page:{},{} ",context.get(START_OFFSET),context.get(END_OFFSET),e);
				
				w=demotionUpsert(upsert,values, context);
				
			}
			return w;
			
		}
		
		
		
		
		
		private int  batchInsert(List<Map<String, Object>> list,Map<String, Object> context) throws WriteErrorOverException {
			
			MapSqlParameterSource[] batch=new MapSqlParameterSource[list.size()];
			int ix=0;
			for (Map<String, Object> data : list) {
				
				MapSqlParameterSource sqlParam=new MapSqlParameterSource();
				
				for (Pair<ColumnInfo, ColumnInfo> pair : columnMappingPairs) {
					
					Object value=data.get(pair.getLeft().getName().toLowerCase()); 
					sqlParam.addValue(pair.getRight().getName(), value, pair.getRight().getJdbcType());
					
				}
				batch[ix]=sqlParam;
				ix++;
			}
			
			try {
				insert.executeBatch(batch);	
				} catch (Exception e) {
				//批量写入错误降级
				log.error(" etl write error,data page:{},{} ",context.get(START_OFFSET),context.get(END_OFFSET),e);
				ix=demotionInsert(batch, insert,context);
				}
			
			return ix;
			
			}

		
		
		/**
		 * 批次失败降级 单条写入，错误数据写入错误日志
		 * @param batch
		 * @param insert
		 * @param context
		 * @throws WriteErrorOverException 
		 */
		private int demotionInsert(MapSqlParameterSource[] batch,SimpleJdbcInsert insert,Map<String, Object> context) throws WriteErrorOverException {
			
			int n=0;
			
			for (MapSqlParameterSource paramsrc : batch) {
				
				try {
					insert.execute(paramsrc);
					n++;
				} catch (Exception e) {
					
					Object idKey=context.get(SRC_PK);
					
					doWriteErrorLog(e, insert.getTableName(),paramsrc.getValues().get(idKey),JSON.toJSONString(paramsrc.getValues()));
					
					log.error(" etl demotionWrite error, error info",e);
					
					Integer errorC=  (Integer) Optional.ofNullable(context.get(ERROR_COUNT)).orElse(0);
					context.put(ERROR_COUNT, ++errorC);
					Integer thresdHold=  (Integer) Optional.ofNullable(context.get(ERROR_THRESHOLD)).orElse(0L);   ;
			
					if(thresdHold>-1 && errorC>thresdHold) {
						throw new WriteErrorOverException((Long) Optional.ofNullable(context.get(context.get(TOTALROWS))).orElse(0L) , thresdHold ,e);
					}
				}	
			}
			return n;
		
		}
		
		private int demotionUpsert(Upsert upsert,List<Object[]> list,Map<String, Object> context) throws WriteErrorOverException{
	
			int n=0;
			
			for (Object[] obj : list) {
				try {
				upsert.updateDirect(obj);
				n++;
				}catch (Exception e) {
					//Object idKey=context.get(SRC_PK);
					//writeErrorLogger.error("errorMsg:{},table:{} ,data:{}",e.getMessage(),upsert.getTableName(),JSON.toJSONString(obj));
					
					doWriteErrorLog(e, upsert.getTableName(),null,JSON.toJSONString(obj));
					
					log.error(" etl demotionWrite error, error info",e);
					Integer errorC=  (Integer) Optional.ofNullable(context.get(ERROR_COUNT)).orElse(0);
					context.put(ERROR_COUNT, ++errorC);
					Integer thresdHold=  (Integer) Optional.ofNullable(context.get(ERROR_THRESHOLD)).orElse(0L);   ;
			
					if(thresdHold>-1 && errorC>thresdHold) {
						throw new WriteErrorOverException((Long) Optional.ofNullable(context.get(context.get(TOTALROWS))).orElse(0L) , thresdHold ,e);
					}
				}
			}
			return n;
			
		}

		public int executeUpdate(String sql,Object[] params) {

			int i=0;
			
			if(params!=null) {
				i=template.update(sql, params);
			}else {
				i=template.update(sql);
			}
			
			return i;
			
		}
		
		
		
		private void doWriteErrorLog(Exception e,String  table,Object pk ,String value)  {
	
			boolean flag=ErrorMessageFilterHelper.getFilter().FilterMessage(e.getMessage());
			
			log.debug("ErrorMessageFilter {} errorMsg:{}, ",flag,e.getMessage());
			
			if(flag) {
				writeErrorLogger.error("table:{} ,Id:{},data:{} ,errorMsg:{}",
						table,pk,JSON.toJSONString(value),e.getMessage() );				
			}
			
		}
		

}
